package com.rin.hdfs;

import com.alibaba.fastjson.JSON;
import com.rin.config.HDFSConfig;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.ReflectionUtils;

import java.io.*;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;

/**
 * @author RinMi
 * @since 2023/6/8
 */

@Service
public class HDFSService {

    private FileSystem fileSystem;

    private final String targetDir = "/Event";

    private final String sourceDir = "D:\\desktop\\temp";

    public HDFSService() {
        System.setProperty("hadoop.home.dir", HDFSConfig.HOOP_HOME_DIR);
        try {
            // 获得FileSystem对象，指定使用root用户上传
            fileSystem = FileSystem.get(new URI(getHdfsUrl()), new Configuration(), HDFSConfig.USERNAME);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 拼接url
     *
     * @return url
     */
    private String getHdfsUrl() {
        StringBuilder builder = new StringBuilder();
        builder.append("hdfs://").append(HDFSConfig.HOSTNAME).append(":").append(HDFSConfig.PORT);
        return builder.toString();
    }

    private String getSerialLocation(String filename) {
        return targetDir + "/" + filename;
    }

    /**
     * 文件上传
     *
     * @param source      资源路径
     * @param destination 存储路径
     */
    public void upload(String source, String destination) {
        try {
            // 创建输入流，参数指定文件输出地址
            InputStream in = new FileInputStream(source);
            // 调用create方法指定文件上传，参数HDFS上传路径
            OutputStream out = fileSystem.create(new Path(destination));
            // 使用Hadoop提供的IOUtils，将in的内容copy到out，设置buffSize大小，是否关闭流设置true
            IOUtils.copyBytes(in, out, 4096, true);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void upload(Object entity, String entityId) {
        try {
            String entityJson = JSON.toJSONString(entity);
            ByteArrayInputStream inputStream = new ByteArrayInputStream(entityJson.getBytes(StandardCharsets.UTF_8));
            // 调用create方法指定文件上传，参数HDFS上传路径
            String destination = "/" + entity.getClass().getSimpleName() + "/" + entityId;
            OutputStream out = fileSystem.create(new Path(destination));
            // 使用Hadoop提供的IOUtils，将in的内容copy到out，设置buffSize大小，是否关闭流设置true
            IOUtils.copyBytes(inputStream, out, 4096, true);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void upload(ByteArrayInputStream inputStream, String destination) {
        try {
            OutputStream out = fileSystem.create(new Path(destination));
            // 使用Hadoop提供的IOUtils，将in的内容copy到out，设置buffSize大小，是否关闭流设置true
            IOUtils.copyBytes(inputStream, out, 4096, true);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 下载文件
     *
     * @param source      资源路径
     * @param destination 存储路径
     */
    public void download(String source, String destination) {
        try {
            // 调用open方法进行下载，参数HDFS路径
            InputStream in = fileSystem.open(new Path(source));
            // 创建输出流，参数指定文件输出地址
            OutputStream out = new FileOutputStream(destination);
            IOUtils.copyBytes(in, out, 4096, true);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public InputStream download(String destination) {
        try {
            // 调用open方法进行下载，参数HDFS路径
            return fileSystem.open(new Path(destination));
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    /**
     * 删除文件
     *
     * @param target 资源路径
     * @return 是否成功
     */
    public boolean delete(String target) {
        boolean flag;
        try {
            // 调用delete方法，删除指定的文件。参数:false:表示是否递归删除
            flag = fileSystem.delete(new Path(target), false);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return flag;
    }

    /**
     * 创建文件夹
     **/
    public boolean mkdir(String directory) {
        boolean flag;
        try {
            // 调用mkdirs方法，在HDFS文件服务器上创建文件夹。
            flag = fileSystem.mkdirs(new Path(directory));
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return flag;
    }

    //==================================================================================//

    public String uploadEntity(InputStream in, String filename) {
        String serialLocation = getSerialLocation(filename);
        try {
            // 调用create方法指定文件上传，参数HDFS上传路径
            OutputStream out = fileSystem.create(new Path(serialLocation));
            // 使用Hadoop提供的IOUtils，将in的内容copy到out，设置buffSize大小，是否关闭流设置true
            IOUtils.copyBytes(in, out, 4096, true);
        } catch (Exception e) {
            serialLocation = null;
            e.printStackTrace();
        }
        return serialLocation;
    }

    /**
     * 文件下载
     **/
    public List<Object> downloadEntity(String serialLocation) {
        try {
            //获取实体类名
            String[] strings = serialLocation.split("/");
            strings = strings[strings.length - 1].split("\\.");
            String entityName = "com.rin.entity." + strings[0];
            //获取数据
            InputStream in = fileSystem.open(new Path(serialLocation));
            //将数据存入实体类中
            return fillBean(entityName, in);
        } catch (Exception e) {
            e.printStackTrace();
            return Collections.emptyList();
        }
    }

    /**
     * 填充实体类数据
     *
     * @param entityName 实体类名称
     * @param in         字节流
     * @return
     * @throws ClassNotFoundException
     * @throws InstantiationException
     * @throws IllegalAccessException
     * @throws InvocationTargetException
     * @throws NoSuchMethodException
     */
    private List<Object> fillBean(String entityName, InputStream in) throws ClassNotFoundException, InstantiationException, IllegalAccessException, InvocationTargetException, NoSuchMethodException {
        //获取实体类class
        Class<?> clazz = Class.forName(entityName);
        //实体类
        Object entity;
        //实体类列表
        List<Object> entityList = new LinkedList<>();
        //获取实体类所有属性
        Field[] declaredFields = clazz.getDeclaredFields();
        InputStreamReader inputStreamReader = new InputStreamReader(in);
        BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
        //暂存每行数据
        String line;
        //获取数据集
        String[] fieldInfos;
        List<Field> fieldList = null;
        int count = 0;
        try {
            while ((line = bufferedReader.readLine()) != null) {
                fieldInfos = line.split(",");
                if (count++ == 0) {
                    //第一次获取字段
                    fieldList = rangeFields(declaredFields, fieldInfos);
                } else {
                    //获取字段数据
                    entity = clazz.getConstructor().newInstance();
                    for (int i = 0; i < declaredFields.length; i++) {
                        Field field = fieldList.get(i);
                        //填充数据
                        transferType(entity, field, fieldInfos[i]);
                    }
                    entityList.add(entity);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            try {
                bufferedReader.close();
                inputStreamReader.close();
                in.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return entityList;
    }

    /**
     * 类型转化、赋值
     *
     * @param object    实体类实例
     * @param field     字段
     * @param fieldInfo 数据
     * @throws IllegalAccessException
     * @throws NoSuchMethodException
     * @throws InstantiationException
     * @throws InvocationTargetException
     */
    private void transferType(Object object, Field field, String fieldInfo) throws IllegalAccessException, NoSuchMethodException, InstantiationException, InvocationTargetException {
        Class<?> type = field.getType();
        ReflectionUtils.makeAccessible(field);
        if (String.class.isAssignableFrom(type)) {
            ReflectionUtils.setField(field, object, fieldInfo);

        } else if (LocalDateTime.class.isAssignableFrom(type)) {
            DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
            LocalDateTime localDateTime = LocalDateTime.parse(fieldInfo, formatter);
            ReflectionUtils.setField(field, object, localDateTime);
        } else {
            Method method = type.getMethod("valueOf", String.class);
            Object typeObject = method.invoke(null, fieldInfo);
            ReflectionUtils.setField(field, object, typeObject);
        }
    }

    /**
     * 按照文本中字段顺序排列属性
     *
     * @param declaredFields 属性数组
     * @param fieldInfos     数据
     * @return 属性列表
     */
    private List<Field> rangeFields(Field[] declaredFields, String[] fieldInfos) {
        List<Field> fieldList = new LinkedList<>();
        for (String fieldInfo : fieldInfos) {
            for (Field declaredField : declaredFields) {
                if (null == fieldInfo || "".equals(fieldInfo)) {
                    throw new RuntimeException("字段名为空");
                }
                if (fieldInfo.equals(declaredField.getName())) {
                    fieldList.add(declaredField);
                }
            }
        }
        return fieldList;
    }
}
